package com.atguigu.day04;

import com.atguigu.utils.IntSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Example1 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .keyBy(r -> 1)
                .process(new MyProcess())
                .print();



        env.execute();
    }

    //输出流上的平均值 截断成整型
    public static class MyProcess extends KeyedProcessFunction<Integer,Integer,Integer>{
        //平均值的状态变量
        private ValueState<Tuple2<Integer,Integer>> sumCount;
        //定时器时间戳 状态变量
        private ValueState<Long>timerTs;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            //初始化平均值状态变量
            sumCount = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple2<Integer, Integer>>("sum-count", Types.TUPLE(Types.INT,Types.INT))
            );

            //初始化定时器
            timerTs = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>("time", Types.LONG)
            );
        }

        @Override
        public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
            if (sumCount.value() == null){
                sumCount.update(Tuple2.of(value,1));
            }else {
                sumCount.update(
                        Tuple2.of(
                                sumCount.value().f0 + value,
                                sumCount.value().f1 + 1
                        )
                );
            }


            if (timerTs.value() == null){
                long currTs = ctx.timerService().currentProcessingTime(); //当前时间
                long tenSecLater = currTs + 10 * 1000L; //十秒之后的时间
                ctx.timerService().registerProcessingTimeTimer(tenSecLater); //注册一个日期
                timerTs.update(tenSecLater); //保存这个注册的时间戳

            }


        }

        //定时器用来向下游发送平均值，满足定义的注册时间10秒才触发一次onTimer
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            out.collect(sumCount.value().f0 / sumCount.value().f1);
            timerTs.clear(); //清除时间戳
        }
    }


}
